import datetime
import os
import platform
import sys
import time
from concurrent.futures import ProcessPoolExecutor, as_completed
from decimal import Decimal

import numpy as np
from loguru import logger

from config import ROOT_DIR, PREMIUM, TRADE_LOG, min_margin_ratio, STRATEGY_LOG

import pandas as pd

from globalVar import GlobalVar
from utils.dataArrange.finalReport import FinalReport
from typing import *
from utils.error import PreparationError
from utils.strategy.conf.BtConf import BtConf
from utils.tools import NullStdout


class FastExecutor:
    """
    快速计算基类
    """

    def __init__(self):
        self.config: Optional[BtConf] = None
        self.async_type: bool = False
        self.dataframe: pd.DataFrame = pd.DataFrame()
        # stdout文件
        self.logfile: Union[TextIO, NullStdout] = NullStdout()

    def calculate(self):
        pass

    def draw(self):
        pass

    @staticmethod
    def _faster(fast: 'FastExecutor', conf: BtConf = None):
        pass

    def run(self, *confs: BtConf) -> None:
        if len(confs) == 0:
            raise PreparationError(f'未输入配置')
        elif len(confs) == 1:
            self._faster(self, confs[0])
            if STRATEGY_LOG:
                self.logfile = open(
                    os.path.join(ROOT_DIR, f'locals', 'output', f'stdout-{self.config.strategy_name}.log'), 'w+',
                    encoding='utf-8')
        else:
            # 多进程时先加载数据存放本地
            confs[0].load_data()
            #
            self.async_type = True
            with ProcessPoolExecutor(max_workers=min(len(confs), os.cpu_count())) as executor:
                result: Dict[str, Dict[str, Any]] = {}
                resp = [executor.submit(self._faster, self, conf) for conf in confs]
                for i, future in enumerate(as_completed(resp)):
                    try:
                        res: Tuple[str, Dict[str, Any]] = future.result()
                        result[res[0]] = res[1]
                    except Exception as e:
                        # raise e
                        logger.error(f'子配置执行错误：{confs[i]}，{e.__str__()}')

            df = pd.DataFrame(result).T
            df.to_csv(os.path.join(ROOT_DIR, 'locals', 'output', f'batch-{confs[0].strategy_name}.csv'))


class CTAExecutor(FastExecutor):
    """
    cta计算，根据signal信号快速计算收益信息并绘图，
    策略中只需要实现pick逻辑，并在结果中添加signal列
    """

    @staticmethod
    def _cal(df: pd.DataFrame, signal_name: str = '', leverage: int = 1, multi: bool = False) -> pd.DataFrame:
        """

        :param df:
        :param signal_name:
        :param leverage:
        :param multi:
        :return:
        """
        signal_col = 'signal' + signal_name
        funding_col = ('_funding_' + signal_name) if multi else '_funding'
        # 去除连续相同信号
        df_filter = df[signal_col].shift(1).fillna(method='ffill') == df[signal_col]
        df.loc[df_filter, signal_col] = np.nan
        # 模拟资金仓位
        df['_pos'] = df[signal_col].shift().fillna(method='ffill').fillna(0)

        # 是否为交易点
        df['_is_trade'] = df['_pos'] != df['_pos'].shift(1).fillna(0)
        # 生成交易序号
        df['_tradeId'] = df.groupby(['_is_trade'])['_is_trade'].rank(method='first')
        df.loc[~df['_is_trade'], '_tradeId'] = None
        df['_tradeId'] = df['_tradeId'].fillna(method='ffill')
        # 下个信号点价格
        df['_priceSell'] = df['close']
        # 仓位结束时间
        df['_timeSell'] = df['open_time']
        df.loc[~df['_is_trade'].shift(-1).fillna(False), '_priceSell'] = None
        df.loc[~df['_is_trade'], '_timeSell'] = None
        df['_priceSell'].fillna(method='bfill', inplace=True)
        df['_timeSell'].fillna(method='bfill', inplace=True)
        df['_timeSell'] = df['_timeSell'].shift(-1)
        # 算出该笔订单的买入价和卖出价格
        df['_priceBuy'] = df['close'].shift(1)
        df.loc[~df['_is_trade'], '_priceBuy'] = np.nan
        df['_priceBuy'] = df['_priceBuy'].fillna(method='ffill')
        # 用最高和最低计算最大盈亏
        df['_lowLine'] = (df['low'] / df['_priceBuy'] - 1) * df['_pos'] * leverage
        df['_highLine'] = (df['high'] / df['_priceBuy'] - 1) * df['_pos'] * leverage
        # 用每个仓位的盈亏量暂时表示资金线
        df[funding_col] = ((df['_priceSell'] / df['_priceBuy'] - 1) * df['_is_trade'] * df['_pos'] * leverage + 1)
        # 将当前仓位盈亏错位到下个仓位起点
        df.loc[~df['_is_trade'], funding_col] = np.nan
        df[funding_col] = df[funding_col].fillna(method='ffill').shift(1)
        df.loc[~df['_is_trade'], funding_col] = 1
        df[funding_col].fillna(1, inplace=True)
        # 根据仓位变化扣除手续费
        df['_trade_volume'] = (df['_pos'] - df['_pos'].shift(1)).fillna(0).abs()
        df['_premium'] = df['_trade_volume'] * float(PREMIUM['swap']['taker']) * leverage
        df[funding_col] *= 1 - df['_premium']
        # 叠乘出实际仓量
        df[funding_col] = df[funding_col].cumprod()
        df[funding_col] *= ((df['close'] / df['_priceBuy'] - 1) * leverage * df['_pos'] + 1)
        df[funding_col].fillna(1, inplace=True)
        # 计算单次仓位收益率
        df['_income'] = (df['_priceSell'] / df['_priceBuy'] - 1) * df['_pos'] * leverage
        df['_income'] = df['_income'].apply(lambda x: -1 if x <= min_margin_ratio - 1 else x)
        # 检测爆仓，小于爆仓线直接置零
        df.loc[(df['_lowLine'] < min_margin_ratio - 1) | (df['_highLine'] < min_margin_ratio - 1), '_funding'] = 0
        # 最早爆仓的交易ID
        min_id: Optional[Union[int, float, Decimal]] = df[df[funding_col] == 0]['_tradeId'].min()
        min_time: Optional[Union[int, float, Decimal]] = df[df[funding_col] == 0]['open_time'].min()

        if min_id is not None:
            df.loc[df['_tradeId'] > min_id, '_is_trade'] = False
            df.loc[df['_tradeId'] > min_id, '_pos'] = 0
            df.loc[df['_tradeId'] >= min_id, '_income'] = -1
            df.loc[df['open_time'] >= min_time, funding_col] = 0

        # df.drop(columns=['_tradeId', '_pos', '_income', '_is_trade', '_lowLine', '_highLine'], inplace=True)
        return df

    def calculate(self):
        """
        计算资金
        :return:
        """
        self.dataframe = self.config.load_data()
        self.dataframe = self.dataframe[self.dataframe["open_time"] >= self.config.s_time.timestamp() * 1000]
        self.dataframe = self.config.strategy.pick(self.dataframe)
        self.dataframe = self._cal(self.dataframe, leverage=self.config.strategy.leverage)

    def draw(self):
        fin = FinalReport('report-' + self.config.strategy_name)
        leverage: int = self.config.strategy.leverage

        self.dataframe.sort_values(by=['open_time', 'symbol'], inplace=True)

        # 资金线
        funding: List[float] = self.dataframe['_funding'].round(4).tolist()

        fin.infos['累计净值'] = round(funding[-1] / funding[0], 2)
        fin.infos['年化'] = round(
            abs(fin.infos['累计净值']) ** (365 / ( self.config.e_time - self.config.s_time).days) - 1, 2)
        fin.infos['月化'] = round((abs(fin.infos['年化']) + 1) ** (1 / 12) - 1, 2) * (
            1 if fin.infos['年化'] >= 0 else -1)
        withdrawals = fin.withdrawal(funding)
        s_time = datetime.datetime.fromtimestamp(self.dataframe["open_time"].values[withdrawals[0]] / 1000,
                                                 tz=datetime.timezone.utc)
        e_time = datetime.datetime.fromtimestamp(self.dataframe["open_time"].values[withdrawals[1]] / 1000,
                                                 tz=datetime.timezone.utc)
        fin.info_table({'最大回撤时间': f'{s_time} - {e_time}', '最大回撤': withdrawals[2]})

        fin.infos['收益回撤比'] = round(fin.infos['年化'] / fin.infos['最大回撤'], 2) if fin.infos[
                                                                                             '最大回撤'] > 0 else '无回撤'

        df = self.dataframe[self.dataframe['_pos'].abs() > 0]
        df = df.groupby('_tradeId', as_index=False).first()

        fin.infos['胜率'] = (str(round(sum(df['_income'] > 0) / len(df), 2) * 100) + '%') if len(df) else '无交易'
        fin.infos['最大盈利'] = str(max(round(df['_income'].max(), 2), 0) * 100) + '%'
        fin.infos['最大亏损'] = str(min(round(df['_income'].min(), 2), 0) * 100) + '%'
        fin.infos['盈亏比'] = -round(
            df[df['_income'] > 0]['_income'].mean() / df[df['_income'] < 0]['_income'].mean(), 2)
        fin.infos['杠杆'] = leverage
        if not self.async_type:
            if TRADE_LOG:
                trade_info: pd.DataFrame = df.loc[:, ['open_time', '_timeSell', 'symbol', '_pos', '_priceBuy',
                                                      '_priceSell', '_trade_volume', '_income', '_premium']]
                trade_info.columns = ['开仓时间', '结仓时间', '币种', '方向', '买入价格', '卖出价格', '交易量', '盈亏',
                                      '手续费']
                trade_info['开仓时间'], trade_info['结仓时间'] = \
                    (pd.to_datetime(trade_info['开仓时间'], unit='ms'),
                     pd.to_datetime(trade_info['结仓时间'], unit='ms'))
                trade_info.to_csv(os.path.join(ROOT_DIR, 'locals/output', f'trade-{self.config.strategy_name}.csv'),
                                  index=False)
            # 时间字符串
            times: List[str] = pd.to_datetime(self.dataframe['open_time'], unit='ms').dt.strftime(
                '%Y-%m-%d %H:%M:%S').tolist()
            fin.account_funding(times, 'funding', funding, self.dataframe['close'].tolist())
            fin.cta_infos(times, funding, self.dataframe, factors=self.config.show_factors)
            fin.generate(os.path.join(ROOT_DIR, 'locals', 'output'))
        return fin.infos

    @staticmethod
    def _faster(fast: FastExecutor, conf: BtConf = None):
        """
        单次计算函数
        :param fast:
        :param conf:
        :return:
        """
        logger.info(f'开始计算：{conf.strategy_name}， 参数：{conf.__str__()}')
        s_time = time.time()
        conf.strategy.injection({'factors': conf.strategy.factorFunc})
        fast.config = conf
        sys.stdout = fast.logfile
        fast.calculate()

        sys.stdout = GlobalVar.BAK_STDOUT
        infos = fast.draw()
        logger.info(f'计算完成：{fast.config.strategy_name}， 参数：{conf.__str__()}，耗时：{int(time.time() - s_time)}s')

        if platform.system() == 'Windows' and not fast.async_type:
            html_f_name = f'report-{fast.config.strategy_name}' + '.html'
            os.system(fr'start {os.path.join(ROOT_DIR, "locals/output", html_f_name)}')
        return fast.config.strategy.system_conf['param'].__str__(), infos
